package com.atguigu.gmall.realtime.utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.java.tuple.Tuple2;
import redis.clients.jedis.Jedis;


import java.util.List;

/**
 * 查询维度数据
 * 优化：旁路缓存
 * 思路：首先先从缓存中获取维度数据，如果缓存中存在维度数据，直接将维度数据返回；
 * 如果说缓存中没有找到对应的维度数据，那么发送请求到Phoenix表中查询维度数据，并将查询维度放到缓存中
 * 分析：type:string
 * 因为我们从维度表中查询维度的数据的时候，是将维度数据封装为了JSONObject，可以直接转换为jsonStr
 * 存到Redis上，这样在从Redis缓存中获取数据的时候，可以很方便的将jsonStr转换为JSONObject
 * expire: 1day
 * 避免一些冷数据常驻内存，给内存带来压力
 * key： dim:维度表表名:18_aaa
 * 注意：如果业务系统维度数据发生了变化，需要将Redis中缓存数据清除掉
 */
public class DimUtil {

    public static JSONObject getDimInfo(String tableName, String id) {
        return getDimInfo(tableName,Tuple2.of("ID",id));
    }

    /**
     * 查询维度数据
     * 优化：旁路缓存
     * 思路：首先先从缓存中获取维度数据，如果缓存中存在维度数据，直接将维度数据返回；
     * 如果说缓存中没有找到对应的维度数据，那么发送请求到Phoenix表中查询维度数据，并将查询维度放到缓存中
     * 分析：type:string
     * 因为我们从维度表中查询维度的数据的时候，是将维度数据封装为了JSONObject，可以直接转换为jsonStr
     * 存到Redis上，这样在从Redis缓存中获取数据的时候，可以很方便的将jsonStr转换为JSONObject
     * expire: 1day
     * 避免一些冷数据常驻内存，给内存带来压力
     * key： dim:维度表表名:18_aaa
     * 注意：如果业务系统维度数据发生了变化，需要将Redis中缓存数据清除掉
     */
    public static JSONObject getDimInfo(String tableName, Tuple2<String, String>... columnNameAndValues) {
        //拼接查询Redis的key
        StringBuilder redisKey = new StringBuilder("dim:" + tableName.toLowerCase() + ":");
        //拼接查询的sql
        StringBuilder selectSql = new StringBuilder("select * from " + tableName + " where ");
        for (int i = 0; i < columnNameAndValues.length; i++) {
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            String columnName = columnNameAndValue.f0;
            String columnValue = columnNameAndValue.f1;
            selectSql.append(columnName + " = '" + columnValue + "'");
            redisKey.append(columnValue);
            if (i < columnNameAndValues.length - 1) {
                selectSql.append(" and ");
                redisKey.append("_");
            }
        }

        //操作Redis客户端对象
        Jedis jedis = null;
        //用于接收从Redis中查询的结果
        String dimInfoStr = null;
        //最终返回的维度对象
        JSONObject dimInfoJsonObj = null;

        try {
            jedis = RedisUtil.getJedis();
            //先从Redis缓存中获取维度数据
            dimInfoStr = jedis.get(redisKey.toString());
        } catch (Exception e) {
            e.printStackTrace();
        }

        //判断是否从Redis中查到了维度数据
        if(dimInfoStr != null && dimInfoStr.length() > 0){
            //在Redis中查到维度数据
            dimInfoJsonObj = JSON.parseObject(dimInfoStr);
        }else{
            //缓存中没有找到对应的维度数据，那么发送请求到Phoenix表中查询维度数据
            System.out.println("从Phoenix表中查询数据的SQL：" + selectSql);
            //底层还是调用PhoenixUtil
            List<JSONObject> jsonObjList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
            if (jsonObjList != null && jsonObjList.size() > 0) {
                dimInfoJsonObj = jsonObjList.get(0);
                //将查询维度放到缓存中  并设置失效时间
                if(jedis != null){
                    jedis.setex(redisKey.toString(),3600*24,dimInfoJsonObj.toJSONString());
                }
            } else {
                System.out.println("没有查到对应的维度数据:" + selectSql);
            }
        }
        if(jedis != null){
            System.out.println("~~关闭Jedis客户端~~");
            jedis.close();
        }
        return dimInfoJsonObj;
    }

    //查询维度数据  为了方便对维度数据进行统一的处理，我们这里将查询结果定义为统一的JSONObject
    public static JSONObject getDimInfoNoCache(String tableName, Tuple2<String, String>... columnNameAndValues) {
        //拼接查询的sql
        StringBuilder selectSql = new StringBuilder("select * from " + tableName + " where ");

        for (int i = 0; i < columnNameAndValues.length; i++) {
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            String columnName = columnNameAndValue.f0;
            String columnValue = columnNameAndValue.f1;
            selectSql.append(columnName + " = '" + columnValue + "'");
            if (i < columnNameAndValues.length - 1) {
                selectSql.append(" and ");
            }
        }
        System.out.println("从Phoenix表中查询数据的SQL：" + selectSql);

        JSONObject dimInfoJsonObj = null;
        //底层还是调用PhoenixUtil
        List<JSONObject> jsonObjList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
        if (jsonObjList != null && jsonObjList.size() > 0) {
            dimInfoJsonObj = jsonObjList.get(0);
        } else {
            System.out.println("没有查到对应的维度数据:" + selectSql);
        }
        return dimInfoJsonObj;
    }

    public static void delCached(String tableName,String id){
        String redisKey = "dim:" + tableName.toLowerCase() + ":" + id;
        try {
            Jedis jedis = RedisUtil.getJedis();
            //删除缓存在Redis中的维度数据
            jedis.del(redisKey);
            jedis.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        //JSONObject dimInfoNoCache = getDimInfoNoCache("dim_base_trademark", Tuple2.of("ID", "18"));
        //JSONObject dimInfoNoCache = getDimInfo("dim_base_trademark", Tuple2.of("ID", "18"));
        JSONObject dimInfoNoCache = getDimInfo("dim_base_trademark", "18");
        System.out.println(dimInfoNoCache);
    }
}
